package com.sojson.util.mq.rabbit.code.test;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Return;
import com.rabbitmq.client.ReturnCallback;
import com.sojson.util.StringUtil;
import com.sojson.util.mq.rabbit.code.RabbitChannelUtil;
import com.sojson.util.mq.rabbit.code.bean.BaseCustomConsumer;
import com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeDelayMsgUtilImpl;
import com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitCodeDelayMsgUtilImpl.RabbitCodeDirectDelayMsgUtilImpl;

import lombok.extern.slf4j.Slf4j;

/**
 * Rabbit消息队列测试类</BR>
 * UI地址http://192.168.100.101:15672</BR>
 * UI默认账号root</BR>
 * UI默认密码123456</BR>
 * 
 * @author liu
 * @date 2021-07-11
 */
@Slf4j
public class RabbitCodeTestUtil {

    /** 交换机名称1 */
    public static final String EXCHANGE_NAME1 = "test1";
    /** 交换机名称2 */
    public static final String EXCHANGE_NAME2 = "test2";

    /** 队列名称1 */
    public static final String QUEUE_NAME1 = "test1";
    /** 队列名称2 */
    public static final String QUEUE_NAME2 = "test2";

    /** 路由Key名称1 */
    public static final String ROUTING_KEY_NAME1 = "test1";
    /** 路由Key名称1 */
    public static final String ROUTING_KEY_NAME2 = "test2";

    /**
     * 每个都确认发送状态</BR>
     * 
     * @param channel
     * @throws Exception
     */
    private static void sc1(Channel channel) throws Exception {
        System.out.println("开始生产消息");

        RabbitCodeDirectDelayMsgUtilImpl instances = RabbitCodeDelayMsgUtilImpl.getDirect();

        // 是否开启发布者确认模式
        boolean isConfirm = true;

        // 是否开启发布者确认模式
        if (isConfirm) {
            // 开启发布者确认模式
            channel.confirmSelect();
        }

        try {
            for (int i = 1; i <= sendNum; i++) {
                // 发送消息
                instances.basicPublish(channel, EXCHANGE_NAME1, StringUtil.EMPTY, (i + "消息"));

                // 确认是否发布成功,如果指定时间(单位:毫秒),则超时后还没有结果就抛异常,如果不指定时间,则一直等到返回成功或失败为止
                // 如果确认为失败,就抛异常

                // 需要调用confirmSelect()方法开启开启发布者确认模式
                // 返回成功还是失败,失败后会继续发送后面的消息(主要用来单独确认)
                // 和waitForConfirmsOrDie()方法不能同时使用,同时使用会导致后面的方法失效
                // 只判断服务器是否收到消息,不能确定消费者是否处理成功
                try {
                    boolean waitForConfirms = channel.waitForConfirms(5000);
                    if (waitForConfirms) {
                        System.out.println(i + ": 发送成功");
                    } else {
                        System.out.println(i + ": 发送失败");
                    }
                } catch (Exception e) {
                    System.out.println(i + ": 超时,发送失败");
                }
            }

            // 需要调用confirmSelect()方法开启开启发布者确认模式
            // 失败后会断开连接并抛出异常,导致不会继续发送后面的消息,已经发送的不会有影响
            // 和waitForConfirms()方法不能同时使用,同时使用会导致后面的方法失效
            // 只判断服务器是否收到消息,不能确定消费者是否处理成功
            // 无法确定哪条消息失败了(主要用来批量确认)
            channel.waitForConfirmsOrDie(5000);
        } finally {
            // 关闭连接,一般消费者不关闭,一直监听队列消息并消费
            RabbitChannelUtil.close(channel, null);
        }
    }

    /**
     * 分批确认发送状态(分批调用回调接口)</BR>
     * 如果一个批次中存在失败的,那这个批次返回的状态就为失败</BR>
     * 
     * @param channel
     * @throws IOException
     */
    private static void sc2(Channel channel) throws IOException {
        System.out.println("开始生产消息");

        RabbitCodeDirectDelayMsgUtilImpl instances = RabbitCodeDelayMsgUtilImpl.getDirect();

        // 是否开启发布者确认模式
        boolean isConfirm = true;

        /**
         * 需要调用confirmSelect()方法开启开启发布者确认模式</BR>
         * 消息发送确认的异步回调</BR>
         * 只判断服务器是否收到消息,不能确定消费者是否处理成功</BR>
         * (!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!)</BR>
         * (!!!!!!!!!!使用这种方式不能关闭连接通道,否则关闭后其他的无法确认!!!!!!!!!!)</BR>
         * (!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!)</BR>
         */
        ConfirmListener confirmListener = new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                // 序列号等于发送顺序
                if (multiple) {
                    System.out.println("批量: 发送失败, 当前序列号: " + deliveryTag);
                } else {
                    System.out.println("单个: 发送失败, 当前序列号: " + deliveryTag);
                }
            }

            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                // 序列号等于发送顺序
                if (multiple) {
                    System.out.println("批量: 发送成功, 当前序列号: " + deliveryTag);
                } else {
                    System.out.println("单个: 发送成功, 当前序列号: " + deliveryTag);
                }
            }
        };

        // 是否开启发布者确认模式
        if (isConfirm) {
            // 开启发布者确认模式
            channel.confirmSelect();

            // 如果有监听器就把监听器添加进去
            if (StringUtil.isNotBlankObject(confirmListener)) {
                // 消息发送成功的异步回调
                channel.addConfirmListener(confirmListener);
            }
        }

        // 故障检测需要在basicPublish()方法上开启故障检测模式
        channel.addReturnListener(new ReturnCallback() {
            @Override
            public void handle(Return returnMessage) {
                System.out.println(
                    "交换机: " + returnMessage.getExchange() + ", 队列: " + returnMessage.getRoutingKey() + ", 不存在!!!");
            }
        });

        for (int i = 1; i <= sendNum; i++) {
            try {
                System.out.println("发送第" + (channel.getNextPublishSeqNo()) + "个");
                // 发送消息
                instances.basicPublish(channel, EXCHANGE_NAME1, StringUtil.EMPTY, true, (i + "消息"));
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private static void xf1(Channel channel) throws IOException {
        RabbitCodeDirectDelayMsgUtilImpl instances = RabbitCodeDelayMsgUtilImpl.getDirect();

        // 初始化消费者
        instances.init(channel);

        // 开始消费消息
        instances.basicConsume(channel, QUEUE_NAME1, new BaseCustomConsumer(channel) {

            /**
            * 收到消息后的处理
            * consumerTag:  创建消息时由channel.basicConsume()传入
            * envelope:     交换机和队列的信息
            * properties:   消息发送时的属性
            * body:         内容
            */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                System.out.println(
                    "交换机: " + envelope.getExchange() + ", 队列: " + envelope.getRoutingKey() + ", 收到消息: " + message);

                try {
                    // 在处理完消息后手动进行确认,如果不确认,默认为返回失败
                    /*
                     * 参数1： 消费消息的index
                     * 参数2： 是否批量确认消息,前提是在同一个channel里面,且是在该消息确认前没有被确认的消息才能批量确认
                     */
                    getChannel().basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception e) {
                    /**
                     * deliveryTag: 消费消息的index
                     * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    channel.basicReject(envelope.getDeliveryTag(), false);
                    /**
                     * deliveryTag: 消费消息的index
                     * multiple: 是否批量否定多个消息,设为false就与basicReject功能一样,true的前提也是在同一个channel,且在该消息否定前存在未确认的消息
                     * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    // channel.basicNack(envelope.getDeliveryTag(), false, true);
                }
            }
        });
    }

    private static void xf2(Channel channel) throws IOException {
        RabbitCodeDirectDelayMsgUtilImpl instances = RabbitCodeDelayMsgUtilImpl.getDirect();

        // 初始化消费者
        instances.init(channel);

        // 开始消费消息
        instances.basicConsume(channel, QUEUE_NAME1, new BaseCustomConsumer(channel) {

            /**
            * 收到消息后的处理
            * consumerTag:  创建消息时由channel.basicConsume()传入
            * envelope:     交换机和队列的信息
            * properties:   消息发送时的属性
            * body:         内容
            */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                System.out.println(
                    "交换机: " + envelope.getExchange() + ", 队列: " + envelope.getRoutingKey() + ", 收到消息: " + message);

                try {
                    // 在处理完消息后手动进行确认,如果不确认,默认为返回失败
                    /*
                     * 参数1： 消费消息的index
                     * 参数2： 是否批量确认消息,前提是在同一个channel里面,且是在该消息确认前没有被确认的消息才能批量确认
                     */
                    getChannel().basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception e) {
                    /**
                     * deliveryTag: 消费消息的index
                     * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    channel.basicReject(envelope.getDeliveryTag(), false);
                    /**
                     * deliveryTag: 消费消息的index
                     * multiple: 是否批量否定多个消息,设为false就与basicReject功能一样,true的前提也是在同一个channel,且在该消息否定前存在未确认的消息
                     * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    // channel.basicNack(envelope.getDeliveryTag(), false, true);
                }
            }
        });
    }

    private static int sendNum = 15;

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitChannelUtil.getChannel();
        RabbitCodeDirectDelayMsgUtilImpl instances = RabbitCodeDelayMsgUtilImpl.getDirect();

        instances.exchangeDeclare(channel, EXCHANGE_NAME1);

        instances.queueDeclare(channel, QUEUE_NAME1);

        instances.queueBind(channel, EXCHANGE_NAME1, QUEUE_NAME1);

        RabbitChannelUtil.close(channel, null);

        xf1(RabbitChannelUtil.getChannel());
        xf2(RabbitChannelUtil.getChannel());
        sc1(RabbitChannelUtil.getChannel());
        sc2(RabbitChannelUtil.getChannel());
    }

}